package com.reger.l2cache.pipeline.pipe;

import java.util.Collection;
//import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
//import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
//import java.util.function.Function;
import java.util.stream.Collectors;

import org.springframework.data.geo.Circle;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.geo.Metric;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
import org.springframework.data.redis.connection.RedisListCommands.Position;
import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
import org.springframework.data.redis.connection.RedisZSetCommands.Range;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.SortParameters;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.lang.Nullable;

import com.reger.l2cache.pipeline.core.Callback;
import com.reger.l2cache.pipeline.core.Pipeline;
import com.reger.l2cache.pipeline.core.PipelineCore;

public class RedisPipeline<K, HK, V> implements Pipeline<K, HK, V> {

	@SuppressWarnings("rawtypes")
	public static final RedisSerializer stringSerializer = new StringRedisSerializer();
	@SuppressWarnings("rawtypes")
	public static final RedisSerializer jdkSerializer = new JdkSerializationRedisSerializer();

	private final Callback<K> kCallback = key -> this.keySerializer.deserialize((byte[]) key);
	@SuppressWarnings("unchecked")
	private final Callback<Set<K>> ksCallback = keys -> {
		if (keys == null) {
			return null;
		}
		return ((Collection<byte[]>) keys).stream().map(bytes -> (K) this.keySerializer.deserialize(bytes))
				.collect(Collectors.toSet());
	};

	private final Callback<V> vCallback = value -> this.valueSerializer.deserialize((byte[]) value);
	@SuppressWarnings("unchecked")
	private final Callback<Set<V>> vSetCallback = values -> {
		if (values == null) {
			return null;
		}
		return ((Collection<byte[]>) values).stream().map(bytes -> (V) this.valueSerializer.deserialize(bytes))
				.collect(Collectors.toSet());
	};
	@SuppressWarnings("unchecked")
	private final Callback<Set<HK>> hkSetCallback = hks -> {
		if (hks == null) {
			return null;
		}
		return ((Collection<byte[]>) hks).stream().map(bytes -> (HK) this.hashKeySerializer.deserialize(bytes))
				.collect(Collectors.toSet());
	};
	@SuppressWarnings("unchecked")
	private final Callback<List<V>> vListCallback = values -> {
		if (values == null) {
			return null;
		}
		return ((Collection<byte[]>) values).stream().map(bytes -> (V) this.valueSerializer.deserialize(bytes))
				.collect(Collectors.toList());
	};

	@SuppressWarnings("unchecked")
	private @Nullable RedisSerializer<K> keySerializer = jdkSerializer;
	@SuppressWarnings("unchecked")
	private @Nullable RedisSerializer<V> valueSerializer = jdkSerializer;
	@SuppressWarnings("unchecked")
	private @Nullable RedisSerializer<HK> hashKeySerializer = jdkSerializer;

	private final PipelineCore pipelineCore;

	public RedisPipeline(PipelineCore pipelineCore) {
		super();
		this.pipelineCore = pipelineCore;
	}

	public RedisSerializer<K> getKeySerializer() {
		return keySerializer;
	}

	public void setKeySerializer(RedisSerializer<K> keySerializer) {
		this.keySerializer = keySerializer;
	}

	public RedisSerializer<V> getValueSerializer() {
		return valueSerializer;
	}

	public void setValueSerializer(RedisSerializer<V> valueSerializer) {
		this.valueSerializer = valueSerializer;
	}

	public RedisSerializer<HK> getHashKeySerializer() {
		return hashKeySerializer;
	}

	public void setHashKeySerializer(RedisSerializer<HK> hashKeySerializer) {
		this.hashKeySerializer = hashKeySerializer;
	}

	@Override
	public <T> T execute(String command, byte[]... args) {
		return pipelineCore.executeNoRelust(c -> c.execute(command, args));
	}

	@Override
	public Boolean exists(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.exists(byteKey));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long del(K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.del(byteKeys));
	}

	@Override
	public DataType type(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.type(byteKey));
	}

	@Override
	public Set<K> keys(String pattern) {
		return pipelineCore.execute(c -> c.keys(pattern.getBytes()), ksCallback);
	}

//	@Override
//	@SuppressWarnings("unchecked")
//	public Cursor<K> scan(ScanOptions options) {
//		return pipelineCore.execute(c -> c.scan(options),cursor->toCursor(s->keySerializer.deserialize(s), (Cursor<byte[]>)cursor));
//	}
	@Override
	public K randomKey() {
		return pipelineCore.execute(c -> c.randomKey(), kCallback);
	}

	@Override
	public void rename(K sourceKey, K targetKey) {
		byte[] byteSourceKey = keySerializer.serialize(sourceKey);
		byte[] byteTargetKey = keySerializer.serialize(targetKey);
		pipelineCore.executeNoRelust(c -> {
			c.rename(byteSourceKey, byteTargetKey);
			return null;
		});
	}

	@Override
	public Boolean renameNX(K sourceKey, K targetKey) {
		byte[] byteSourceKey = keySerializer.serialize(sourceKey);
		byte[] byteTargetKey = keySerializer.serialize(targetKey);
		return pipelineCore.execute(c -> c.renameNX(byteSourceKey, byteTargetKey));
	}

	@Override
	public Boolean expire(K key, long seconds) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.expire(byteKey, seconds));
	}

	@Override
	public Boolean pExpire(K key, long millis) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.pExpire(byteKey, millis));
	}

	@Override
	public Boolean expireAt(K key, long unixTime) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.expireAt(byteKey, unixTime));
	}

	@Override
	public Boolean pExpireAt(K key, long unixTimeInMillis) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.pExpireAt(byteKey, unixTimeInMillis));
	}

	@Override
	public Boolean persist(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.persist(byteKey));
	}

	@Override
	public Boolean move(K key, int dbIndex) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.move(byteKey, dbIndex));
	}

	@Override
	public Long ttl(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.ttl(byteKey));
	}

	@Override
	public Long ttl(K key, TimeUnit timeUnit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.ttl(byteKey, timeUnit));
	}

	@Override
	public Long pTtl(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.pTtl(byteKey));
	}

	@Override
	public Long pTtl(K key, TimeUnit timeUnit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.pTtl(byteKey, timeUnit));
	}

	@Override
	public List<Object> sort(K key, SortParameters params) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sort(byteKey, params));
	}

	@Override
	public Long sort(K key, SortParameters params, K storeKey) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteStoreKey = keySerializer.serialize(storeKey);
		return pipelineCore.execute(c -> c.sort(byteKey, params, byteStoreKey));
	}

	@Override
	public byte[] dump(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.dump(byteKey));
	}

	@Override
	public void restore(K key, long ttlInMillis, byte[] serializedValue) {
		byte[] byteKey = keySerializer.serialize(key);
		pipelineCore.executeNoRelust(c -> {
			c.restore(byteKey, ttlInMillis, serializedValue);
			return null;
		});
	}

	@Override
	public V get(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.get(byteKey), vCallback);
	}

	@Override
	public V getSet(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.getSet(byteKey, byteValue), vCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public List<V> mGet(K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.mGet(byteKeys), vListCallback);
	}

	@Override
	public Boolean set(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.set(byteKey, byteValue));
	}

	@Override
	public Boolean set(K key, V value, Expiration expiration, SetOption option) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.set(byteKey, byteValue, expiration, option));
	}

	@Override
	public Boolean setNX(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.setNX(byteKey, byteValue));
	}

	@Override
	public Boolean setEx(K key, long seconds, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.setEx(byteKey, seconds, byteValue));
	}

	@Override
	public Boolean setNxEx(K key, long seconds, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.execute("set", byteKey, byteValue, "EX".getBytes(),
				String.valueOf(seconds).getBytes(), "NX".getBytes()), ret -> "OK".equals(ret));
	}

	@Override
	public Boolean pSetEx(K key, long milliseconds, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.pSetEx(byteKey, milliseconds, byteValue));
	}

	@Override
	public Boolean pSetNxEx(K key, long milliseconds, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.execute("set", byteKey, byteValue, "PX".getBytes(),
				String.valueOf(milliseconds).getBytes(), "NX".getBytes()), ret -> "OK".equals(ret));
	}

	@Override
	public Boolean mSet(Map<K, V> tuple) {
		Map<byte[], byte[]> byteTuple = new HashMap<>(tuple.size());
		tuple.forEach((k, v) -> byteTuple.put(keySerializer.serialize(k), valueSerializer.serialize(v)));
		return pipelineCore.execute(c -> c.mSet(byteTuple));
	}

	@Override
	public Boolean mSetNX(Map<K, V> tuple) {
		Map<byte[], byte[]> byteTuple = new HashMap<>(tuple.size());
		tuple.forEach((k, v) -> byteTuple.put(keySerializer.serialize(k), valueSerializer.serialize(v)));
		return pipelineCore.execute(c -> c.mSetNX(byteTuple));
	}

	@Override
	public Long incr(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.incr(byteKey));
	}

	@Override
	public Long incrBy(K key, long value) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.incrBy(byteKey, value));
	}

	@Override
	public Double incrBy(K key, double value) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.incrBy(byteKey, value));
	}

	@Override
	public Long decr(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.decr(byteKey));
	}

	@Override
	public Long decrBy(K key, long value) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.decrBy(byteKey, value));
	}

	@Override
	public Long append(K key, byte[] value) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.append(byteKey, value));
	}

	@Override
	public byte[] getRange(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.getRange(byteKey, start, end));
	}

	@Override
	public void setRange(K key, byte[] value, long offset) {
		byte[] byteKey = keySerializer.serialize(key);
		pipelineCore.executeNoRelust(c -> {
			c.setRange(byteKey, value, offset);
			return null;
		});
	}

	@Override
	public Boolean getBit(K key, long offset) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.getBit(byteKey, offset));
	}

	@Override
	public Boolean setBit(K key, long offset, boolean value) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.setBit(byteKey, offset, value));
	}

	@Override
	public Long bitCount(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.bitCount(byteKey));
	}

	@Override
	public Long bitCount(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.bitCount(byteKey, start, end));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long bitOp(BitOperation op, K destKey, K... keys) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.bitOp(op, byteDestKey, byteKeys));
	}

	@Override
	public Long strLen(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.strLen(byteKey));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long rPush(K key, V... values) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteValuess = new byte[values.length][];
		for (int i = 0; i < values.length; i++) {
			byteValuess[i] = valueSerializer.serialize(values[i]);
		}
		return pipelineCore.execute(c -> c.rPush(byteKey, byteValuess));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long lPush(K key, V... values) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteValuess = new byte[values.length][];
		for (int i = 0; i < values.length; i++) {
			byteValuess[i] = valueSerializer.serialize(values[i]);
		}
		return pipelineCore.execute(c -> c.lPush(byteKey, byteValuess));
	}

	@Override
	public Long rPushX(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValues = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.rPushX(byteKey, byteValues));
	}

	@Override
	public Long lPushX(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValues = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.lPushX(byteKey, byteValues));
	}

	@Override
	public Long lLen(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.lLen(byteKey));
	}

	@Override
	public List<V> lRange(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.lRange(byteKey, start, end), vListCallback);
	}

	@Override
	public void lTrim(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		pipelineCore.executeNoRelust(c -> {
			c.lTrim(byteKey, start, end);
			return null;
		});
	}

	@Override
	public V lIndex(K key, long index) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.lIndex(byteKey, index), vCallback);
	}

	@Override
	public Long lInsert(K key, Position where, V pivot, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] bytePivot = valueSerializer.serialize(pivot);
		byte[] byteValues = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.lInsert(byteKey, where, bytePivot, byteValues));
	}

	@Override
	public void lSet(K key, long index, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValues = valueSerializer.serialize(value);
		pipelineCore.executeNoRelust(c -> {
			c.lSet(byteKey, index, byteValues);
			return null;
		});
	}

	@Override
	public Long lRem(K key, long count, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValues = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.lRem(byteKey, count, byteValues));
	}

	@Override
	public V lPop(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.lPop(byteKey), vCallback);
	}

	@Override
	public V rPop(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.rPop(byteKey), vCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public List<V> bLPop(int timeout, K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.bLPop(timeout, byteKeys), vListCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public List<V> bRPop(int timeout, K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.bRPop(timeout, byteKeys), vListCallback);
	}

	@Override
	public V rPopLPush(K srcKey, K dstKey) {
		byte[] byteSrcKey = keySerializer.serialize(srcKey);
		byte[] byteDestKey = keySerializer.serialize(dstKey);
		return pipelineCore.execute(c -> c.rPopLPush(byteSrcKey, byteDestKey), vCallback);
	}

	@Override
	public V bRPopLPush(int timeout, K srcKey, K destKey) {
		byte[] byteSrcKey = keySerializer.serialize(srcKey);
		byte[] byteDestKey = keySerializer.serialize(destKey);
		return pipelineCore.execute(c -> c.bRPopLPush(timeout, byteSrcKey, byteDestKey), vCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long sAdd(K key, V... values) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteValuess = new byte[values.length][];
		for (int i = 0; i < values.length; i++) {
			byteValuess[i] = valueSerializer.serialize(values[i]);
		}
		return pipelineCore.execute(c -> c.sAdd(byteKey, byteValuess));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long sRem(K key, V... values) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteValuess = new byte[values.length][];
		for (int i = 0; i < values.length; i++) {
			byteValuess[i] = valueSerializer.serialize(values[i]);
		}
		return pipelineCore.execute(c -> c.sRem(byteKey, byteValuess));
	}

	@Override
	public V sPop(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sPop(byteKey), vCallback);
	}

	@Override
	public List<V> sPop(K key, long count) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sPop(byteKey, count), vListCallback);
	}

	@Override
	public Boolean sMove(K srcKey, K destKey, V value) {
		byte[] byteSrcKey = keySerializer.serialize(srcKey);
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.sMove(byteSrcKey, byteDestKey, byteValue));
	}

	@Override
	public Long sCard(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sCard(byteKey));
	}

	@Override
	public Boolean sIsMember(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.sIsMember(byteKey, byteValue));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Set<V> sInter(K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.sInter(byteKeys), vSetCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long sInterStore(K destKey, K... keys) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.sInterStore(byteDestKey, byteKeys));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Set<V> sUnion(K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.sUnion(byteKeys), vs -> {
			if (vs == null) {
				return null;
			}
			return ((List<byte[]>) vs).stream().map(bytes -> (V) valueSerializer.deserialize(bytes))
					.collect(Collectors.toSet());
		});
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long sUnionStore(K destKey, K... keys) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.sUnionStore(byteDestKey, byteKeys));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Set<V> sDiff(K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.sDiff(byteKeys), vSetCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long sDiffStore(K destKey, K... keys) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.sDiffStore(byteDestKey, byteKeys));
	}

	@Override
	public Set<V> sMembers(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sMembers(byteKey), vSetCallback);
	}

	@Override
	public V sRandMember(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sRandMember(byteKey), vCallback);
	}

	@Override
	public List<V> sRandMember(K key, long count) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.sRandMember(byteKey, count), vListCallback);
	}

//	@Override
//	@SuppressWarnings("unchecked")
//	public Cursor<V> sScan(K key, ScanOptions options) {
//		byte[] byteKey=keySerializer.serialize(key);
//		return pipelineCore.execute(c -> c.sScan(byteKey, options),cursor->toCursor(s->valueSerializer.deserialize(s), (Cursor<byte[]>)cursor));
//	}

	@Override
	public Boolean zAdd(K key, double score, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.zAdd(byteKey, score, byteValue));
	}

	@Override
	public Long zAdd(K key, Set<Tuple> tuples) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zAdd(byteKey, tuples));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long zRem(K key, V... values) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteValues = new byte[values.length][];
		for (int i = 0; i < byteValues.length; i++) {
			byteValues[i] = valueSerializer.serialize(values[i]);
		}
		return pipelineCore.execute(c -> c.zRem(byteKey, byteValues));
	}

	@Override
	public Double zIncrBy(K key, double increment, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.zIncrBy(byteKey, increment, byteValue));
	}

	@Override
	public Long zRank(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.zRank(byteKey, byteValue));
	}

	@Override
	public Long zRevRank(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.zRevRank(byteKey, byteValue));
	}

	@Override
	public Set<V> zRange(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRange(byteKey, start, end), vSetCallback);
	}

	@Override
	public Set<Tuple> zRangeWithScores(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRangeWithScores(byteKey, start, end));
	}

	@Override
	public Set<Tuple> zRangeByScoreWithScores(K key, Range range, Limit limit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRangeByScoreWithScores(byteKey, range, limit));
	}

	@Override
	public Set<V> zRevRange(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRevRange(byteKey, start, end), vSetCallback);
	}

	@Override
	public Set<Tuple> zRevRangeWithScores(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRevRangeWithScores(byteKey, start, end));
	}

	@Override
	public Set<V> zRevRangeByScore(K key, Range range, Limit limit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRevRangeByScore(byteKey, range, limit), vSetCallback);
	}

	@Override
	public Set<Tuple> zRevRangeByScoreWithScores(K key, Range range, Limit limit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRevRangeByScoreWithScores(byteKey, range, limit));
	}

	@Override
	public Long zCount(K key, Range range) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zCount(byteKey, range));
	}

	@Override
	public Long zCard(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zCard(byteKey));
	}

	@Override
	public Double zScore(K key, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.zScore(byteKey, byteValue));
	}

	@Override
	public Long zRemRange(K key, long start, long end) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRemRange(byteKey, start, end));
	}

	@Override
	public Long zRemRangeByScore(K key, Range range) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRemRangeByScore(byteKey, range));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long zUnionStore(K destKey, K... sets) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteSets = new byte[sets.length][];
		for (int i = 0; i < byteSets.length; i++) {
			byteSets[i] = keySerializer.serialize(sets[i]);
		}
		return pipelineCore.execute(c -> c.zUnionStore(byteDestKey, byteSets));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long zUnionStore(K destKey, Aggregate aggregate, int[] weights, K... sets) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteSets = new byte[sets.length][];
		for (int i = 0; i < byteSets.length; i++) {
			byteSets[i] = keySerializer.serialize(sets[i]);
		}
		return pipelineCore.execute(c -> c.zUnionStore(byteDestKey, aggregate, weights, byteSets));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long zInterStore(K destKey, K... sets) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteSets = new byte[sets.length][];
		for (int i = 0; i < byteSets.length; i++) {
			byteSets[i] = keySerializer.serialize(sets[i]);
		}
		return pipelineCore.execute(c -> c.zInterStore(byteDestKey, byteSets));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long zInterStore(K destKey, Aggregate aggregate, int[] weights, K... sets) {
		byte[] byteDestKey = keySerializer.serialize(destKey);
		byte[][] byteSets = new byte[sets.length][];
		for (int i = 0; i < byteSets.length; i++) {
			byteSets[i] = keySerializer.serialize(sets[i]);
		}
		return pipelineCore.execute(c -> c.zInterStore(byteDestKey, aggregate, weights, byteSets));
	}

	@Override
	public Cursor<Tuple> zScan(K key, ScanOptions options) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zScan(byteKey, options));
	}

	@Override
	public Set<V> zRangeByScore(K key, String min, String max, long offset, long count) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRangeByScore(byteKey, min, max, offset, count));
	}

	@Override
	public Set<V> zRangeByScore(K key, Range range, Limit limit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRangeByScore(byteKey, range, limit), vSetCallback);
	}

	@Override
	public Set<V> zRangeByLex(K key, Range range, Limit limit) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.zRangeByLex(byteKey, range, limit), vSetCallback);
	}

	@Override
	public Boolean hSet(K key, HK field, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteField = hashKeySerializer.serialize(field);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.hSet(byteKey, byteField, byteValue));
	}

	@Override
	public Boolean hSetNX(K key, HK field, V value) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteField = hashKeySerializer.serialize(field);
		byte[] byteValue = valueSerializer.serialize(value);
		return pipelineCore.execute(c -> c.hSetNX(byteKey, byteField, byteValue));
	}

	@Override
	public V hGet(K key, HK field) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteField = hashKeySerializer.serialize(field);
		return pipelineCore.execute(c -> c.hGet(byteKey, byteField), vCallback);
	}

	@Override
	public List<V> hMGet(K key, @SuppressWarnings("unchecked") HK... fields) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteFields = new byte[fields.length][];
		for (int i = 0; i < byteFields.length; i++) {
			byteFields[i] = hashKeySerializer.serialize(fields[i]);
		}
		return pipelineCore.execute(c -> c.hMGet(byteKey, byteFields), vListCallback);
	}

	@Override
	public void hMSet(K key, Map<HK, V> hashes) {
		byte[] byteKey = keySerializer.serialize(key);
		Map<byte[], byte[]> byteHashes = new HashMap<>(hashes.size());
		hashes.forEach((k, v) -> byteHashes.put(hashKeySerializer.serialize(k), valueSerializer.serialize(v)));
		pipelineCore.executeNoRelust(c -> {
			c.hMSet(byteKey, byteHashes);
			return null;
		});
	}

	@Override
	public Long hIncrBy(K key, HK field, long delta) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteField = hashKeySerializer.serialize(field);
		return pipelineCore.execute(c -> c.hIncrBy(byteKey, byteField, delta));
	}

	@Override
	public Double hIncrBy(K key, HK field, double delta) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteField = hashKeySerializer.serialize(field);
		return pipelineCore.execute(c -> c.hIncrBy(byteKey, byteField, delta));
	}

	@Override
	public Boolean hExists(K key, HK field) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteField = hashKeySerializer.serialize(field);
		return pipelineCore.execute(c -> c.hExists(byteKey, byteField));
	}

	@Override
	@SuppressWarnings("unchecked")
	public Long hDel(K key, HK... fields) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteFields = new byte[fields.length][];
		for (int i = 0; i < byteFields.length; i++) {
			byteFields[i] = hashKeySerializer.serialize(fields[i]);
		}
		return pipelineCore.execute(c -> c.hDel(byteKey, byteFields));
	}

	@Override
	public Long hLen(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.hLen(byteKey));
	}

	@Override
	public Set<HK> hKeys(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.hKeys(byteKey), hkSetCallback);
	}

	@Override
	public List<V> hVals(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.hVals(byteKey), vListCallback);
	}

	@Override
	@SuppressWarnings("unchecked")
	public Map<HK, V> hGetAll(K key) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.hGetAll(byteKey), maps -> {
			if (maps == null) {
				return null;
			}
			Map<HK, V> rets = new HashMap<>(((Map<byte[], byte[]>) maps).size());
			((Map<byte[], byte[]>) maps)
					.forEach((k, v) -> rets.put(hashKeySerializer.deserialize(k), valueSerializer.deserialize(v)));
			return rets;
		});
	}

//	@Override
//	@SuppressWarnings("unchecked")
//	public Cursor<Entry<HK, V>> hScan(K key, ScanOptions options) {
//		byte[] byteKey=keySerializer.serialize(key);
//		return  pipelineCore.execute(c -> c.hScan(byteKey, options),cursor->toCursor(s-> new Entry<HK, V>(){
//				@Override
//				public HK getKey() {return hashKeySerializer.deserialize(s.getKey()); }
//				@Override
//				public V getValue() {return valueSerializer.deserialize(s.getValue());}
//				@Override
//				public V setValue(V value) {return valueSerializer.deserialize(s.setValue(valueSerializer.serialize(value)));}			
//			}, (Cursor<Entry<byte[],byte[]>>)cursor));
//	}

	@Override
	public void scriptFlush() {
		pipelineCore.executeNoRelust(c -> {
			c.scriptFlush();
			return null;
		});
	}

	@Override
	public void scriptKill() {
		pipelineCore.executeNoRelust(c -> {
			c.scriptKill();
			return null;
		});
	}

	@Override
	public String scriptLoad(String script) {
		return pipelineCore.executeNoRelust(c -> c.scriptLoad(script.getBytes()));
	}

	@Override
	public List<Boolean> scriptExists(String... scriptShas) {
		return pipelineCore.executeNoRelust(c -> c.scriptExists(scriptShas));
	}

	@Override
	public <T> T eval(String script, ReturnType returnType, int numKeys, byte[]... keysAndArgs) {
		return pipelineCore.execute(c -> c.eval(script.getBytes(), returnType, numKeys, keysAndArgs));
	}

	@Override
	public <T> T evalSha(String scriptSha, ReturnType returnType, int numKeys, byte[]... keysAndArgs) {
		return pipelineCore.executeNoRelust(c -> c.evalSha(scriptSha, returnType, numKeys, keysAndArgs));
	}

	@Override
	public Long geoAdd(K key, Point point, V member) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteMember = valueSerializer.serialize(member);
		return pipelineCore.execute(c -> c.geoAdd(byteKey, point, byteMember));
	}

	@Override
	public Long geoAdd(K key, Map<V, Point> memberCoordinateMap) {
		byte[] byteKey = keySerializer.serialize(key);
		Map<byte[], Point> memberBytesCoordinateMap = new HashMap<>();
		memberCoordinateMap.forEach((k, p) -> memberBytesCoordinateMap.put(valueSerializer.serialize(k), p));
		return pipelineCore.execute(c -> c.geoAdd(byteKey, memberBytesCoordinateMap));
	}

	@Override
	public Long geoAdd(K key, Iterable<GeoLocation<V>> locations) {
		byte[] byteKey = keySerializer.serialize(key);
		List<GeoLocation<byte[]>> byteLocations = new LinkedList<>();
		locations.forEach(
				t -> byteLocations.add(new GeoLocation<byte[]>(valueSerializer.serialize(t.getName()), t.getPoint())));
		return pipelineCore.execute(c -> c.geoAdd(byteKey, byteLocations));
	}

	@Override
	public Distance geoDist(K key, V member1, V member2) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteMember1 = valueSerializer.serialize(member1);
		byte[] byteMember2 = valueSerializer.serialize(member2);
		return pipelineCore.execute(c -> c.geoDist(byteKey, byteMember1, byteMember2));
	}

	@Override
	public Distance geoDist(K key, V member1, V member2, Metric metric) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteMember1 = valueSerializer.serialize(member1);
		byte[] byteMember2 = valueSerializer.serialize(member2);
		return pipelineCore.execute(c -> c.geoDist(byteKey, byteMember1, byteMember2, metric));
	}

	@Override
	public List<String> geoHash(K key, @SuppressWarnings("unchecked") V... members) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteMembers = new byte[members.length][];
		for (int i = 0; i < byteMembers.length; i++) {
			byteMembers[i] = valueSerializer.serialize(members[i]);
		}
		return pipelineCore.execute(c -> c.geoHash(byteKey, byteMembers));
	}

	@Override
	public List<Point> geoPos(K key, @SuppressWarnings("unchecked") V... members) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteMembers = new byte[members.length][];
		for (int i = 0; i < byteMembers.length; i++) {
			byteMembers[i] = valueSerializer.serialize(members[i]);
		}
		return pipelineCore.execute(c -> c.geoPos(byteKey, byteMembers));
	}

	@Override
	public GeoResults<GeoLocation<V>> geoRadius(K key, Circle within) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.geoRadius(byteKey, within));
	}

	@Override
	public GeoResults<GeoLocation<V>> geoRadius(K key, Circle within, GeoRadiusCommandArgs args) {
		byte[] byteKey = keySerializer.serialize(key);
		return pipelineCore.execute(c -> c.geoRadius(byteKey, within, args));
	}

	@Override
	public GeoResults<GeoLocation<V>> geoRadiusByMember(K key, V member, Distance radius) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteMember = valueSerializer.serialize(member);
		return pipelineCore.execute(c -> c.geoRadiusByMember(byteKey, byteMember, radius));
	}

	@Override
	public GeoResults<GeoLocation<V>> geoRadiusByMember(K key, V member, Distance radius, GeoRadiusCommandArgs args) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[] byteMember = valueSerializer.serialize(member);
		return pipelineCore.execute(c -> c.geoRadiusByMember(byteKey, byteMember, radius, args));
	}

	@Override
	public Long geoRemove(K key, @SuppressWarnings("unchecked") V... members) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteMembers = new byte[members.length][];
		for (int i = 0; i < byteMembers.length; i++) {
			byteMembers[i] = valueSerializer.serialize(members[i]);
		}
		return pipelineCore.execute(c -> c.geoRemove(byteKey, byteMembers));
	}

	@Override
	public Long pfAdd(K key, @SuppressWarnings("unchecked") V... values) {
		byte[] byteKey = keySerializer.serialize(key);
		byte[][] byteValues = new byte[values.length][];
		for (int i = 0; i < byteValues.length; i++) {
			byteValues[i] = valueSerializer.serialize(values[i]);
		}
		return pipelineCore.execute(c -> c.pfAdd(byteKey, byteValues));
	}

	@Override
	public Long pfCount(@SuppressWarnings("unchecked") K... keys) {
		byte[][] byteKeys = new byte[keys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(keys[i]);
		}
		return pipelineCore.execute(c -> c.pfCount(byteKeys));
	}

	@Override
	public void pfMerge(K destinationKey, @SuppressWarnings("unchecked") K... sourceKeys) {
		byte[] byteDestKey = keySerializer.serialize(destinationKey);
		byte[][] byteKeys = new byte[sourceKeys.length][];
		for (int i = 0; i < byteKeys.length; i++) {
			byteKeys[i] = keySerializer.serialize(sourceKeys[i]);
		}
		pipelineCore.executeNoRelust(c -> {
			c.pfMerge(byteDestKey, byteKeys);
			return null;
		});
	}

}
